use std::path::PathBuf;
use std::sync::Arc;

use parking_lot::RwLock;
use prost::bytes::{Buf, Bytes, BytesMut};

use crate::data::log_record::LogRecordType;
use crate::fio::{self, new_io_manager};

use crate::errors::{self, Result};

use super::log_record::{LogRecord, ReadLogRecord};

pub const DATA_FILE_NAME_SUFFIX:&str = ".data";

pub struct DataFile {
    file_id: Arc<RwLock<u32>>,
    write_off: Arc<RwLock<u64>>,
    io_manager: Box<dyn fio::IoManager>,
}

impl DataFile {
    // 创建或者打开一个新的数据文件
    pub fn new(path: PathBuf, file_id: u32) -> Result<DataFile> {
        // 根据path和id 构建完整文件名称
        let file_name = get_data_file_name(path,file_id.clone());
        // 初始化IO manager
        let io_manager = new_io_manager(file_name)?;
        
        Ok(
            DataFile {
                file_id: Arc::new(RwLock::new(file_id)),
                write_off: Arc::new(RwLock::new(0)),
                io_manager: Box::new(io_manager),
            }
        )
    }

    pub fn get_write_off(&self) -> u64 {
        let read_guard = self.write_off.read();
        *read_guard
    }

    pub fn set_write_off(&self, write_off: u64) {
        let mut write_guard = self.write_off.write();
        *write_guard = write_off;
    }

    pub fn get_file_id(&self) -> u32 {
        let read_guard = self.file_id.read();
        *read_guard
    }

    // 根据偏移量从数据文件读取数据并生成结构体信息
    pub fn read_log_record(&self, off: u64) -> Result<ReadLogRecord> {
        // 数据模型 type（1字节）+ keysize（4字节u32）+ valuesize(8字节u64) + 用户存储的key
        // + 用户存储的value + CRC校验值（4字节u32）
        let mut header_buf = BytesMut::zeroed(13);

        self.io_manager.read(&mut header_buf, off)?;

        let rec_type = header_buf.get_u8();

        let keysize = header_buf.get_u32() as usize;

        let valuesize = header_buf.get_u64() as usize;

        if keysize ==0 && valuesize==0 {
            return Err(errors::Errors::ReadDataEOFError);
        }

        // 读取实际key、value、crc 数据
        let mut kv_buf = BytesMut::zeroed(keysize  + valuesize  + 4);

        self.io_manager.read(&mut kv_buf, off + 13)?;

        let mut log_record = LogRecord {
            key: kv_buf.get(..keysize).unwrap().to_vec(),
            value: kv_buf.get(keysize..kv_buf.len()-4).unwrap().to_vec(),
            rec_type:LogRecordType::from_u8(rec_type),
        };

        // 向前移动到最后4个字节，就是crc
        kv_buf.advance(keysize+valuesize);

        if kv_buf.get_u32() != log_record.get_crc() {
            return  Err(errors::Errors::InvalidLogRecordCrc);
        }

        Ok(ReadLogRecord{
            record: log_record,
            size: (13 + keysize + valuesize + 4) as u64,
        })
        
    }

    pub fn sync(&self) -> Result<()> {
        self.io_manager.sync()
    }

    pub fn write(&self, buf: &[u8]) -> Result<usize> {
        let n_bytes = self.io_manager.write(buf)?;

        let mut write_off = self.write_off.write();
        *write_off += n_bytes as u64;

        Ok(n_bytes)

    }
}


// 获取文件名称
fn get_data_file_name(path:PathBuf,file_id:u32) -> PathBuf {
    let v = std::format!("{:09}",file_id) + DATA_FILE_NAME_SUFFIX;
    path.join(v)
}

#[cfg(test)]
mod test {
    use crate::data::data_file;

    use super::*;

    #[test]
    fn test_new_data_file(){
        let dir_path = std::env::temp_dir();
        let data_file_res = DataFile::new(dir_path.clone(), 0);
        assert!(data_file_res.is_ok());

        let data_file = data_file_res.unwrap();
        assert_eq!(data_file.get_file_id(),0);

        println!("temp dir:{:?}",dir_path.clone().as_os_str());
    }

    #[test]
    fn test_data_file_write(){
        let dir_path = std::env::temp_dir();
        let data_file_res = DataFile::new(dir_path.clone(), 0);
        assert!(data_file_res.is_ok());
        let data_file = data_file_res.unwrap();
        assert_eq!(data_file.get_file_id(),0);

        let res = data_file.write("hello,my kv db!".as_bytes());
        assert!(res.is_ok());

    }

    #[test]
    fn test_data_file_sync(){
        let dir_path = std::env::temp_dir();
        let data_file_res = DataFile::new(dir_path.clone(), 1);
        assert!(data_file_res.is_ok());
        let data_file = data_file_res.unwrap();
        assert_eq!(data_file.get_file_id(),1);

        let res = data_file.sync();
        assert!(res.is_ok());

    }


}